package com.xahj.bd2104.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}

/**
 * Created with IntelliJ IDEA.
 * Author: Amos
 * E-mail: amos@amoscloud.com
 * Date: 2021/9/13
 * Time: 9:47
 * Description: 
 */
object SparkCoreDemo4_缓存操作 {
  def main(args: Array[String]): Unit = {
    // 4. 缓存操作  缓存  检查点
    //   cache persist
    //   checkpoint

    val conf = new SparkConf()
      .setAppName("Demo4")
      .setMaster("local[2]")
    val sc = new SparkContext(conf)

    val rdd: RDD[Int] = sc.makeRDD(1 to 10)

    //  4.1 cache
    //  底层调用空参的persist
    rdd.cache()

    // 4.2 persist
    // 底层调用 传入存储级别为 MEMORY_ONLY 的persist
    //persist(StorageLevel.MEMORY_ONLY)
    rdd.persist()

    // 4.3 有参的persist
    // 可以根据需要选择各种不同的存储级别
    //  常用的存储级别
    //   StorageLevel.MEMORY_ONLY  仅内存：在内存充足的情况下使用  如果内存不足 会造成溢出
    //   StorageLevel.MEMORY_AND_DISK  优先使用内存， 内存不足将数据持久化在磁盘
    //   StorageLevel.DISK_ONLY 仅磁盘 明确知道内存不足，或者任务的计算时间要求较低
    //   StorageLevel.OFF_HEAP 使用非堆(java堆内存) 存储RDD中的元素对象 可以避免JVM的GC损耗
    //                      非堆 安全性较差  建议在私有集群使用
//    rdd.persist()


    //4.4 创建检查点

    // cache 和persist都是临时存储
    // 数据会随任务结束而自动删除
    // 如果想将RDD的数据永久存储 可以使用checkpoint机制

    // 4.4.1 设置检查点目录
    sc.setCheckpointDir("C:\\Users\\Amos\\Desktop\\checkpoint")
    // 4.4.2 创建检查点
    rdd.checkpoint()

    rdd.foreach(println)
  }
}
